package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.sink.HBaseSinkFunction;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.Random;

/**
 * @Author lzc
 * @Date 2023/7/4 13:59
 */
public class FlinkSinkUtil {
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getHBaseSink() {
        return new HBaseSinkFunction();
    }
    
    public static Sink<String> getKafkaSink(String topic) {
        
        return KafkaSink.<String>builder()
            .setBootstrapServers(Constant.KAFKA_BROKERS)
            .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                                     .setTopic(topic)
                                     .setValueSerializationSchema(new SimpleStringSchema())
                                     .build())
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix("atguigu-" + topic + new Random().nextLong())
            .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
            .build();
    }
    
    public static Sink<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        return KafkaSink.<Tuple2<JSONObject, TableProcess>>builder()
            .setBootstrapServers(Constant.KAFKA_BROKERS)
            .setRecordSerializer(new KafkaRecordSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Nullable
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> t,
                                                                KafkaSinkContext ctx,
                                                                Long timestamp) {
                    String topic = t.f1.getSinkTable();
                    JSONObject data = t.f0;
                    data.remove("op_type");
                    return new ProducerRecord<>(topic, data.toJSONString().getBytes(StandardCharsets.UTF_8));
                }
            })
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix("atguigu-" + new Random().nextLong())
            .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
            .build();
    }
    
    public static DorisSink<String> getDorisSink(String table) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true"); // 每行一条 json 数据
        return DorisSink.<String>builder()
            .setDorisReadOptions(DorisReadOptions.builder().build())
            .setDorisOptions(DorisOptions.builder()
                                 .setFenodes(Constant.DORIS_FE_NODES)
                                 .setTableIdentifier(table)
                                 .setUsername("root")
                                 .setPassword("aaaaaa")
                                 .build()
            )
            .setDorisExecutionOptions(DorisExecutionOptions.builder()
                                          .setLabelPrefix("a")
                                          .disable2PC()  // 禁用两阶段提交: 如果不禁用LabelPrefix必须全局唯一
                                          .setBufferCount(3)
                                          .setBufferSize(1024 * 1024)
                                          .setCheckInterval(3000)
                                          .setMaxRetries(3)
                                          .setStreamLoadProp(props)
                                          .build())
            .setSerializer(new SimpleStringSerializer())
            .build();
        
    }
}
